package repo

import (
	"git.multiverse.io/eventkit/common/util"
	"git.multiverse.io/eventkit/dxc-server/models"
	"git.multiverse.io/framework/log"
	"github.com/pkg/errors"
	"sync"
	"time"
)

const (
	ShardCount                           = 32
	CannotFoundTransactionErrorFormatter = "transaction{RootXid:[%s]} cannot be found!"
)

// @Desc a struct to keep RootTransaction by Memory
type ConcurrentMapShared struct {
	items map[string]*models.RootTransaction
	index int
	sync.RWMutex
}

// @Desc check memory storage shared
// for TTL
//func (c *ConcurrentMapShared) run() {
//	log.Infof("memory storage shared[shared index:%d] time expired[%d seconds] checker running...", c.index, config.DtsAgentCfg.ExpireSeconds)
//	t := time.NewTicker(time.Second * 5)
//	for {
//		select {
//		case <-t.C:
//			c.Lock()
//			for k, v := range c.items {
//				if "0" == v.RootTxnStat && (time.Now().Unix()-v.RootTxnRegisterTime.Unix()) > config.DtsAgentCfg.ExpireSeconds {
//					log.Errorf("transaction{Shard index:[%d], RootXid:[%s]} time expired[%d]s, start discard...", c.index, v.RootXid, config.DtsAgentCfg.ExpireSeconds)
//					delete(c.items, k)
//					log.Errorf("discarded transaction{RootXid:[%s]} successfully", v.RootXid)
//				}
//			}
//			c.Unlock()
//		}
//	}
//}

// @Desc get memory storage shared
func (m ConcurrentMap) GetShard(key string) *ConcurrentMapShared {
	return m[uint(util.Fnv32(key))%uint(ShardCount)]
}

type ConcurrentMap []*ConcurrentMapShared

// memory transaction repository start
type MemoryTxnRepository struct {
	//storage map[string]*models.RootTransaction
	//sync.RWMutex
	shardStorage ConcurrentMap
}

func NewMemoryTxnRepository() *MemoryTxnRepository {
	log.Infos("Start initialize memory repository...")
	defer log.Infos("Successfully initialized memory repository!")

	memoryTxnRepository := &MemoryTxnRepository{}
	memoryTxnRepository.shardStorage = make(ConcurrentMap, ShardCount)
	for i := 0; i < ShardCount; i++ {
		memoryTxnRepository.shardStorage[i] = &ConcurrentMapShared{
			items: make(map[string]*models.RootTransaction),
			index: i,
		}
	}

	return memoryTxnRepository
}

func (m *MemoryTxnRepository) AddRoot(transaction *models.Transaction) (result int, err error) {
	// shard map
	sm := m.shardStorage.GetShard(transaction.RootXid)

	sm.Lock()
	defer sm.Unlock()

	_, ok := sm.items[transaction.RootXid]
	if ok {
		err = errors.Errorf("transaction{RootXid:[%s]} is already exists!", transaction.RootXid)
		return 0, err
	}
	// add root transaction
	branches := make([]*models.Transaction, 0)
	branches = append(branches, transaction)
	rootTransaction := models.RootTransaction{
		RootXid:             transaction.RootXid,
		RootTxnStat:         "0",
		RootTxnRegisterTime: time.Now(),
		Branches:            branches,
	}
	sm.items[transaction.RootXid] = &rootTransaction
	return 1, nil
}

func (m *MemoryTxnRepository) AddBranch(transaction *models.Transaction) (result int, err error) {
	// shard map
	sm := m.shardStorage.GetShard(transaction.RootXid)

	sm.Lock()
	defer sm.Unlock()

	root, ok := sm.items[transaction.RootXid]
	if !ok {
		err = errors.Errorf("transaction{RootXid:[%s]} not exists!", transaction.RootXid)
		return 0, err
	}
	for _, txn := range root.Branches {
		if txn.RootXid == transaction.RootXid && txn.BranchXid == transaction.BranchXid {
			err = errors.Errorf("transaction{RootXid:[%s],BranchXid:[%s]} is already exists!", transaction.RootXid, transaction.BranchXid)
			return 0, err
		}
	}
	root.Branches = append(root.Branches, transaction)
	return 1, nil
}

// @Desc implements MemoryTxnRepository's Add method to save transaction information
// @Param transaction transaction information
// @Return result the number of saved transaction
// @Return err error
func (m *MemoryTxnRepository) Add(transaction *models.Transaction) (result int, err error) {
	// shard map
	sm := m.shardStorage.GetShard(transaction.RootXid)

	sm.Lock()
	defer sm.Unlock()

	root, ok := sm.items[transaction.RootXid]

	if !ok {
		branches := make([]*models.Transaction, 0)
		branches = append(branches, transaction)
		// root transaction
		rootTransaction := models.RootTransaction{
			RootXid:             transaction.RootXid,
			RootTxnStat:         "0",
			RootTxnRegisterTime: time.Now(),
			Branches:            branches,
		}
		sm.items[transaction.RootXid] = &rootTransaction
	} else {
		for _, txn := range root.Branches {
			if txn.RootXid == transaction.RootXid && txn.BranchXid == transaction.BranchXid {
				err = errors.Errorf("transaction{RootXid:[%s],BranchXid:[%s]} is already exists!", transaction.RootXid, transaction.BranchXid)
				return 0, err
			}
		}
		root.Branches = append(root.Branches, transaction)
	}

	return 1, nil
}

func (m *MemoryTxnRepository) Recovery(root *models.RootTransaction) (int, error) {
	sm := m.shardStorage.GetShard(root.RootXid)
	sm.Lock()
	defer sm.Unlock()

	_, ok := sm.items[root.RootXid]
	if ok {
		return 0, nil
	}
	sm.items[root.RootXid] = root
	return 1, nil
}

// @Desc implements MemoryTxnRepository's FindAllBranchByRootXid method to find all branch transaction information by rootXid
// @Param rootXid transaction's rootXid
// @Param rootXid transaction information
// @Return result the number of saved transaction
// @Return err error
func (m *MemoryTxnRepository) FindAllBranchByRootXid(rootXid string) (result *models.RootTransaction, err error) {
	// shard map
	sm := m.shardStorage.GetShard(rootXid)

	sm.RLock()
	defer sm.RUnlock()

	if root, ok := sm.items[rootXid]; !ok {
		err = errors.Errorf(CannotFoundTransactionErrorFormatter, rootXid)
		return
	} else {
		return root, nil
	}
}

// @Desc implements MemoryTxnRepository's DeleteTxnByXid method to delete transaction by rootXid and branchXid
// @Param rootXid transaction's rootXid
// @Param branchXid transaction's branchXid
// @Return result the number of deleted transaction
// @Return err error
func (m *MemoryTxnRepository) DeleteTxnByXid(rootXid string, branchXid string) (result int, err error) {
	// shard map
	sm := m.shardStorage.GetShard(rootXid)

	sm.Lock()
	defer sm.Unlock()

	if root, ok := sm.items[rootXid]; !ok {
		err = errors.Errorf(CannotFoundTransactionErrorFormatter, rootXid)
		return 0, err
	} else {
		if rootXid == branchXid {
			branchNums := len(root.Branches)
			delete(sm.items, rootXid)
			return branchNums, nil
		}

		for i := 0; i < len(root.Branches); i++ {
			if (root.Branches)[i].BranchXid == branchXid {
				return 1, nil
			}
		}
	}

	err = errors.Errorf("transaction{RootXid:[%s],BranchXid:[%s]} cannot be found!", rootXid, branchXid)
	return
}

func (m *MemoryTxnRepository) DeleteTxn(rootXid string) (result int, err error) {
	// shard map
	sm := m.shardStorage.GetShard(rootXid)

	sm.Lock()
	defer sm.Unlock()

	if root, ok := sm.items[rootXid]; !ok {
		err = errors.Errorf(CannotFoundTransactionErrorFormatter, rootXid)
		return 0, err
	} else {
		branchNums := len(root.Branches)
		delete(sm.items, rootXid)
		return branchNums, nil
	}
}

// @Desc implements MemoryTxnRepository's IsExistsRootTransaction method to check the root transaction exists or not
// @Param rootXid transaction's rootXid
// @Return isExists root transaction exists or not
// @Return err error
func (m *MemoryTxnRepository) IsExistsRootTransaction(rootXid string) (isExists bool, err error) {
	// shard map
	sm := m.shardStorage.GetShard(rootXid)

	sm.RLock()
	defer sm.RUnlock()

	_, isExists = sm.items[rootXid]

	return
}

func (m *MemoryTxnRepository) IsExistsBranchTransaction(rootXid string, branchXid string) (bool, error) {
	sm := m.shardStorage.GetShard(rootXid)
	sm.RLock()
	defer sm.RUnlock()

	root, ok := sm.items[rootXid]
	if !ok {
		return false, errors.Errorf(CannotFoundTransactionErrorFormatter, rootXid)
	}
	for _, branch := range root.Branches {
		if branch.BranchXid == branchXid {
			return true, nil
		}
	}
	return false, nil
}

// @Desc implements MemoryTxnRepository's TryCallback method to change root transaction stat
// @Param rootXid transaction's rootXid
// @Param callbackType callback typecode
// @Return isExists root transaction exists or not
// @Return err error
func (m *MemoryTxnRepository) TryCallback(rootXid string, callbackType string) (result *models.RootTransaction, err error) {
	// shard map
	sm := m.shardStorage.GetShard(rootXid)

	sm.Lock()
	defer sm.Unlock()

	root, ok := sm.items[rootXid]
	if !ok {
		return nil, errors.Errorf("TryCallback failed, Cannot found transaction with RootXid:[%s]", rootXid)
	} else {
		if "0" != root.RootTxnStat {
			return nil, errors.Errorf("TryCallback failed, Invalid root transaction stat[%s]", root.RootTxnStat)
		} else {
			root.RootTxnStat = callbackType
		}
	}
	return root, nil
}

// @Desc implements MemoryTxnRepository's TryEnList method to check root transaction stat
// @Param rootXid transaction's rootXid
// @Return err error
func (m *MemoryTxnRepository) TryEnList(rootXid string) (err error) {
	// shard map
	sm := m.shardStorage.GetShard(rootXid)

	sm.RLock()
	defer sm.RUnlock()

	root, ok := sm.items[rootXid]
	if !ok {
		return errors.Errorf("TryEnList failed, Cannot found transaction with RootXid:[%s]", rootXid)
	} else {
		if "0" != root.RootTxnStat {
			return errors.Errorf("TryEnList failed, Invalid root transaction stat[%s]", root.RootTxnStat)
		}
	}
	return nil
}

// update branch RecoveryTimes、BranchTxnStat、BranchTxnLastUpdateTime
func (m *MemoryTxnRepository) CallbackFail(rootXid string, branches []*models.Transaction, firstTimeCallback bool) (int, error) {
	mp := make(map[string]*models.Transaction)
	for _, branch := range branches {
		mp[branch.BranchXid] = branch
	}
	sm := m.shardStorage.GetShard(rootXid)

	sm.Lock()
	defer sm.Unlock()
	root, ok := sm.items[rootXid]
	if !ok {
		return 0, errors.Errorf("CallbackFail failed, Cannot found transaction with RootXid:[%s]", rootXid)
	}
	index := 0
	if firstTimeCallback {
		root.RootTxnStartRecoverTime = time.Now()
		index++
	}
	for _, branch := range root.Branches {
		_, ok := mp[branch.BranchXid]
		if ok {
			index++
			branch.RecoveryTimes = mp[branch.BranchXid].RecoveryTimes
			branch.BranchTxnStat = mp[branch.BranchXid].BranchTxnStat
			branch.BranchTxnLastUpdateTime = mp[branch.BranchXid].BranchTxnLastUpdateTime
		}
	}
	return index, nil
}

func (m *MemoryTxnRepository) UpdateBranchTryFailedIgnoreCallbackCancel(rootXid string, branchXid string) (int, error) {
	sm := m.shardStorage.GetShard(rootXid)

	sm.RLock()
	defer sm.RUnlock()
	root, ok := sm.items[rootXid]
	if !ok {
		return 0, errors.Errorf("UpdateBranchTryFailedIgnoreCallbackCancel failed, Cannot found transaction with RootXid:[%s]", rootXid)
	}
	for _, branch := range root.Branches {
		if branch.BranchXid == branchXid {
			branch.TryFailedIgnoreCallbackCancel = true
			return 1, nil
		}
	}
	return 0, errors.Errorf("UpdateBranchTryFailedIgnoreCallbackCancel failed, Cannot found transaction with BranchXid:[%s]", branchXid)
}

// 获取注册时间超时的根事务详细信息
func (m *MemoryTxnRepository) FindRootBranchesByTimeExpire(expireSeconds int64) ([]*models.Transaction, error) {
	result := make([]*models.Transaction, 0)
	now := time.Now()
	for _, c := range m.shardStorage {
		c.RLock()
		for _, v := range c.items {
			if now.Sub(v.RootTxnRegisterTime).Seconds() > float64(expireSeconds) && !v.HasExpireTime {
				result = append(result, v.Branches[0])
			}
		}
		c.RUnlock()
	}
	return result, nil
}

func (m *MemoryTxnRepository) ExpireTime(rootXid string) (int, error) {
	sm := m.shardStorage.GetShard(rootXid)

	sm.RLock()
	defer sm.RUnlock()
	root, ok := sm.items[rootXid]
	if !ok {
		return 0, errors.Errorf("ExpireTime failed, Cannot found transaction with RootXid:[%s]", rootXid)
	}
	root.HasExpireTime = true
	return 1, nil
}

func (m *MemoryTxnRepository) LoadRootsFromPersistenceRepo(count int64) ([]string, error) {
	panic("MemoryTxnRepository not support")
}
